package com.shujia.spark.mllib

import org.apache.spark.ml.linalg
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.regression.{LinearRegression, LinearRegressionModel}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo2Point {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("point")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    /**
     * 1、读取原始数据
     *
     */
    val pointsDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("label DOUBLE,featuresStr STRING")
      .load("data/points.txt")

    //将一个特征字符串转换成特征向量
    val toVector: UserDefinedFunction = udf((featuresStr: String) => {
      val xs: Array[Double] = featuresStr
        .split("\\|")
        .map(_.toDouble)

      //将x转换成向量返回
      Vectors.dense(xs)
    })

    val trainDF: DataFrame = pointsDF
      //将特征字符串转换成特征向量
      .select($"label", toVector($"featuresStr") as "features")


    /**
     * 选择算法，不同的机器学习问题选择不同的算法
     * 分类：y的值是离散的
     * 回归：y的值是连续的
     * 线性回归，岭回归
     *
     */

    //构建算法
    val lr = new LinearRegression()

    /**
     * 将训练数据带入算法训练模型
     *
     * 底层使用spark进行分布式计算，训练模型
     *
     * 模型：k和b加上公式
     *
     */
    val model: LinearRegressionModel = lr.fit(trainDF)


    //获取结局和斜率
    println(model.intercept)
    println(model.coefficients)

    /**
     * 来了一个新的x,可以预测y
     *
     */

    val y: Double = model.predict(Vectors.dense(Array(100.0)))
    println(y)

  }

}
